package dws

import modes.{DimAreaInfo, OrderPriceInfo}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

object AreaOrder {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val dimAreaSrc: DataStream[(String, String)] = env.addSource(new DimAreaHBaseReader)
    val orderSrc: DataStream[(String, String)] = env.addSource(new OrderHBaseReader)

    val dimArea: DataStream[DimAreaInfo] = dimAreaSrc.map(x => {
      val areaId: String = x._1
      val items: Array[String] = x._2.split("-")
      DimAreaInfo(areaId.toInt, items(0), items(1).toInt, items(2), items(3).toInt, items(4))
    })

    val orders: DataStream[OrderPriceInfo] = orderSrc.map(x => {
      val orderId: String = x._1
      val items: Array[String] = x._2.split(",")
      val areaId: String = items(0)
      val totalMoney: String = items(12)
      OrderPriceInfo(orderId.toInt, areaId.toInt, totalMoney.toDouble)
    })

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    tEnv.createTemporaryView("dimArea", dimArea)
    tEnv.createTemporaryView("orders", orders)

    val sql: String =
      """
        |select cityName, provinceName, sum(totalMoney) amount, count(*) cnt
        |from orders o
        |inner join dimArea d on o.areaId = d.areaId
        |group by cityName, provinceName
        |""".stripMargin

    val resultTable: Table = tEnv.sqlQuery(sql)
    val result: DataStream[(Boolean, Row)] = tEnv.toRetractStream[Row](resultTable)

    result.print()
    env.execute()

  }
}
